package org.example.mq;

/**
 * @FileName CustomerRocketMQConsumer
 * @Description
 * @Author coderfun
 * @selfDesc coding in fun
 * @date 2024-11-07
 **/

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.example.lua.LuaService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;


@Component
public class CustomerRocketMQConsumer {
    private static final Logger logger = LoggerFactory.getLogger(CustomerRocketMQConsumer.class);
    @Autowired
    private LuaService luaService;
    @Value("${rocketmq.name-server}")
    private String nameServerAddr;
    @Value("${rocketmq.consumer.consumeMessageBatchMaxSize}")
    private int consumeMessageBatchMaxSize;
    @Value("${rocketmq.consumer.limitConsumerSize}")
    private int limitConsumerSize;


    @Bean
    public DefaultMQPushConsumer defaultMQPushConsumer() {
        try {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("counter-consumer");
            consumer.setNamesrvAddr(nameServerAddr);
            consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
            consumer.subscribe("counter", "*");
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                System.out.println("Received message count: " + msgs.size());
                if (msgs.size() > consumeMessageBatchMaxSize) {
                    //进行写聚合
                    logger.info("准备写聚合");
                } else {
                    //单步执行Redis指令
                    for (MessageExt msg : msgs) {
                        logger.info("开始执行Redis指令");
                        System.out.println("Received message: " + new String(msg.getBody()));
                        // 在这里处理每条消息
                        String s = new String(msg.getBody());
                        String substring = s.substring(1, s.length() - 1);
                        String[] split = substring.split(" ");
                        luaService.hIncrByWithLua(split[1], split[2], Long.parseLong(split[3]));
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
            System.out.println("Consumer Started.");
            return consumer;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }
}
